package com.wunong.smart.bigdata.flink.function;

import com.wunong.smart.bigdata.flink.data.AnalysisData;
import com.wunong.smart.bigdata.flink.data.SourceData;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

/**
 * @author created by zealot.zt
 */
public class ExcludeValueFunction extends RichCoFlatMapFunction<SourceData, AnalysisData, AnalysisData> {

    /**
     * 输入数据
     */
    private ValueState<SourceData> soureState;

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<SourceData> descriptor = new ValueStateDescriptor<>("inputState", SourceData.class);
        soureState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap1(SourceData controlValue, Collector<AnalysisData> out) throws Exception {
        // 发起连接的流：若流1存在，则设置值
        soureState.update(controlValue);
    }

    @Override
    public void flatMap2(AnalysisData dataValue, Collector<AnalysisData> out) throws Exception {
        // 被连接的流：若流1中存在，则添加
        SourceData sourceData = soureState.value();

        if (sourceData != null) {
            // 回填数据
            dataValue.setSourceData(sourceData);
            out.collect(dataValue);
        }
    }

}